package studio.raptor.databus.client;

import com.google.protobuf.InvalidProtocolBufferException;
import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.command.ActiveMQMessage;
import studio.raptor.databus.command.Event;

import javax.jms.*;

/**
 * 消费者对象
 */
public final class DatabusConsumer {

    private Connection connection;
    private Session session ;
    private MessageConsumer messageConsumer;
    private static Logger log = LoggerFactory.getLogger(DatabusConsumer.class);
    private DatabusConfig databusConfig ;
    private Integer eventSeq;
    private String eventName ;

    /**
     * 消费者构造方法-对包内其他类开放
     *
     */
    DatabusConsumer(){}


    Connection getConnection() {
        return connection;
    }

    void setConnection(Connection connection) {
        this.connection = connection;
    }

    Session getSession() {
        return session;
    }

    void setSession(Session session) {
        this.session = session;
    }

    MessageConsumer getMessageConsumer() {
        return messageConsumer;
    }

    void setMessageConsumer(MessageConsumer messageConsumer) {
        this.messageConsumer = messageConsumer;
    }

    DatabusConfig getDatabusConfig() {
        return databusConfig;
    }

    void setDatabusConfig(DatabusConfig databusConfig) {
        this.databusConfig = databusConfig;
    }

    Integer getEventSeq() {
        return eventSeq;
    }

    void setEventSeq(Integer eventSeq) {
        this.eventSeq = eventSeq;
    }

    String getEventName() {
        return eventName;
    }

    void setEventName(String eventName) {
        this.eventName = eventName;
    }


    /**
     * 设置监听器
     * @param messageListener 消息监听器
     * @throws JMSException
     */
    public void setMessageListener(MessageListener messageListener) throws JMSException {
        this.messageConsumer.setMessageListener(messageListener);
    }

    /**
     * 获取消息。若Broker中无消息，会一直阻塞
     * @return
     * @throws JMSException
     */
    public Event receive() throws JMSException, InvalidProtocolBufferException {
        if( !checkRecieveAuthorization(this.getEventSeq(),this.getEventName()) ){
            throw new JMSException("您无权订阅事件:["+eventName+"-"+eventSeq+"],若需订阅请先申请");
        }
        //如何保证在这个方法中获得到当前消费者的目的地。
        ActiveMQMessage mqMessage = (ActiveMQMessage)this.messageConsumer.receive();
        EventMessage eventMessage =
            new EventMessage(mqMessage,this.getDatabusConfig().TRANSPORTTYPE);
        Event event = eventMessage.getEvent() ;
        return event;
    }


    /**
     * 获取消息，不设置获取消息的超时时间
     * @return 若无消息则返回NULL
     * @throws JMSException
     */
    public Event receive(long receiveTimeOut) throws JMSException, InvalidProtocolBufferException {
        if( !checkRecieveAuthorization(this.getEventSeq(),this.getEventName()) ){
            throw new JMSException("您无权订阅事件:["+eventName+"-"+eventSeq+"],若需订阅请先申请");
        }
        ActiveMQMessage mqMessage = (ActiveMQMessage)this.messageConsumer.receive(receiveTimeOut);
        EventMessage eventMessage =
            new EventMessage(mqMessage,this.getDatabusConfig().TRANSPORTTYPE);
        Event event = eventMessage.getEvent() ;
        return event;
    }

    /**
     * 获取消息，不设置获取消息的超时时间
     * @return 若无消息则返回NULL
     * @throws JMSException
     */
    public Event receiveNoWait() throws JMSException, InvalidProtocolBufferException {
        if( !checkRecieveAuthorization(this.getEventSeq(),this.getEventName()) ){
            throw new JMSException("您无权订阅事件:["+eventName+"-"+eventSeq+"],若需订阅请先申请");
        }
        Event event = null ;
        ActiveMQMessage message = (ActiveMQMessage)this.messageConsumer.receiveNoWait();
        EventMessage eventMessage = new EventMessage(message,this.getDatabusConfig().TRANSPORTTYPE);
        event = eventMessage.getEvent();
        return event;
    }

    /**
     * 获取尚未ACK过的消息。使用此方法所获得的消息，需要手动进行ACK，如:eventMessage.acknowledge()
     * @param receiveTimeout 接受消息超时时间，此值小于1直接返回NULL
     * @return 返回EventMessage 消息，若为空表示队列为空
     * @throws JMSException
     */
    public EventMessage receiveUnACKMessage(long receiveTimeout) throws JMSException {
        if( !checkRecieveAuthorization(this.getEventSeq(),this.getEventName()) ){
            throw new JMSException("您无权订阅事件:["+eventName+"-"+eventSeq+"],若需订阅请先申请");
        }
        //校验是否有权限接受
        if( receiveTimeout <1L ){
            return null;
        }
        ActiveMQMessage activeMQMessage = (ActiveMQMessage)this.messageConsumer.receive(receiveTimeout);
        if(Objects.equals(null,activeMQMessage) ){
            return null;
        }
        EventMessage eventMessage = new EventMessage(activeMQMessage,this.getDatabusConfig().TRANSPORTTYPE);
        return eventMessage ;
    }

    /**
     * 关闭消费者，关闭后不可再用此消费者者接受消息
     */
    public void close()  {
        try {
            if (this.messageConsumer != null) {
                this.messageConsumer.close();
            }
            if (this.session != null) {
                this.session.close();
            }
            if (this.connection != null) {
                this.connection.close();
            }
        } catch (JMSException e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 通知服务端重新投递当前的消息，重新投递次数由配置控制<br>
     * 服务端重新投递消息到客户端需要一定耗时，因此当前方法执行完毕后，直接执行获取消息方法可能还是会获取为空，建议执行完此方法后等待一小段时间
     * @throws JMSException
     */
    public void notifyRedeliver() throws JMSException {
        this.session.recover();
    }


    /**
     * <p>校验当前消费者是否有权接受指定事件，当且仅当eventKindId和eventName都满足时校验通过</p>
     * @param eventKindId 事件序号
     * @param eventName 事件名称
     * @return
     *
     * @deprecated As of databus-client version 0.1.3-SNAPSHOT,
     * replaced by <code>checkRecieveAuthorization(Integer eventKindId,String eventName)</code>.
     */
    @Deprecated
    public boolean receiveAuthorizationCheck(Integer eventKindId,String eventName){

        return checkRecieveAuthorization(eventKindId,eventName);
    }

    /**
     * 检查待发送事件是否被授权发送
     * @param eventKindId 事件kindId
     * @param eventName 事件名称
     * @return
     */
    public boolean checkRecieveAuthorization(Integer eventKindId,String eventName){
        return databusConfig.getBizPubSubEventInfo().getSub().containsKey(eventName+"_"+eventKindId);
    }


}
